FastAPI এবং MongoDB Integration

FastAPI এবং MongoDB-কে একসাথে ব্যবহার করা খুবই কার্যকরী এবং স্কেলেবল অ্যাপ্লিকেশন তৈরি করতে সাহায্য করে। MongoDB হলো একটি NoSQL ডাটাবেস যা ডকুমেন্ট-বেসড স্টোরেজ ব্যবস্থাপনা ব্যবহার করে। FastAPI-তে MongoDB ইন্টিগ্রেশন করার জন্য আমরা motor লাইব্রেরি ব্যবহার করতে পারি, যা একটি asynchronous MongoDB ড্রাইভার।

এখানে আমরা দেখব কিভাবে FastAPI এবং MongoDB ইন্টিগ্রেট করা যায় এবং একটি সিম্পল CRUD (Create, Read, Update, Delete) অ্যাপ তৈরি করা যায়।

Step 1: MongoDB ইনস্টল করা

MongoDB আপনার সিস্টেমে ইনস্টল থাকতে হবে অথবা আপনি MongoDB Atlas ব্যবহার করতে পারেন, যা একটি মেঘ ভিত্তিক MongoDB সার্ভিস।

আপনার সিস্টেমে MongoDB ইনস্টল না থাকলে, MongoDB ডাউনলোড এবং ইনস্টল করার জন্য অফিসিয়াল ডকুমেন্টেশন অনুসরণ করুন।

Step 2: FastAPI এবং MongoDB লাইব্রেরি ইনস্টল করা

MongoDB এর সাথে কাজ করার জন্য FastAPI তে motor ড্রাইভারটি ব্যবহার করা হয়। motor হল একটি অ্যাসিঙ্ক্রোনাস MongoDB ড্রাইভার যা asyncio-এর সাথে কাজ করে।

pip install fastapi
pip install motor
pip install uvicorn

Step 3: FastAPI এবং MongoDB ইন্টিগ্রেশন

এখন, আমরা FastAPI এবং MongoDB ইন্টিগ্রেট করব। MongoDB সংযোগ স্থাপন এবং CRUD অপারেশন করার জন্য motor ব্যবহার করা হবে।

উদাহরণ: FastAPI এবং MongoDB Integration

from fastapi import FastAPI, HTTPException
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel
from typing import List

app = FastAPI()

# MongoDB URI
MONGO_URI = "mongodb://localhost:27017"  # আপনার MongoDB URI এখানে দিন
DATABASE_NAME = "mydatabase"
COLLECTION_NAME = "items"

# MongoDB ডাটাবেসে সংযোগ স্থাপন
client = AsyncIOMotorClient(MONGO_URI)
db = client[DATABASE_NAME]
collection = db[COLLECTION_NAME]

# Pydantic Model
class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float | None = None

@app.on_event("startup")
async def startup_db():
    """MongoDB এ সংযোগ স্থাপন"""
    global client, db, collection
    client = AsyncIOMotorClient(MONGO_URI)
    db = client[DATABASE_NAME]
    collection = db[COLLECTION_NAME]

@app.on_event("shutdown")
async def shutdown_db():
    """MongoDB সংযোগ বন্ধ করা"""
    client.close()

# Create Item (POST)
@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    result = await collection.insert_one(item_dict)
    return {"item_id": str(result.inserted_id)}

# Get all items (GET)
@app.get("/items/", response_model=List[Item])
async def get_items():
    items = []
    async for item in collection.find():
        items.append(Item(**item))
    return items

# Get item by id (GET)
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: str):
    item = await collection.find_one({"_id": item_id})
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return Item(**item)

# Update item (PUT)
@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
    update_result = await collection.update_one(
        {"_id": item_id},
        {"$set": item.dict(exclude_unset=True)}
    )
    if update_result.matched_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item updated successfully"}

# Delete item (DELETE)
@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    delete_result = await collection.delete_one({"_id": item_id})
    if delete_result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

Step 4: কোডের ব্যাখ্যা

1. MongoDB Connection:
   • MongoDB সার্ভারের সাথে সংযোগ স্থাপন করতে AsyncIOMotorClient ব্যবহার করা হয়েছে।
   • client এর মাধ্যমে MongoDB ডাটাবেস এবং কালেকশন অ্যাক্সেস করা হয়।
2. Pydantic Model:
   • Item মডেলটি FastAPI রিকোয়েস্ট এবং রেসপন্স ভ্যালিডেশন এবং ডাটা টাইপ যাচাইয়ের জন্য ব্যবহার করা হয়েছে।
3. CRUD অপারেশন:
   • Create: POST /items/ - MongoDB-তে একটি নতুন আইটেম তৈরি করে।
   • Read:
     • GET /items/ - MongoDB কালেকশন থেকে সমস্ত আইটেম ফেরত দেয়।
     • GET /items/{item_id} - নির্দিষ্ট আইটেম item_id দ্বারা MongoDB থেকে পড়ে।
   • Update: PUT /items/{item_id} - আইটেম আপডেট করার জন্য MongoDB-তে সংরক্ষণ করে।
   • Delete: DELETE /items/{item_id} - আইটেম মুছে ফেলে MongoDB থেকে।
4. AsyncIO:
   • async এবং await ব্যবহার করে Motor এর সাথে MongoDB তে asynchronously অপারেশন সম্পাদন করা হয়েছে।
   • async for ব্যবহার করে MongoDB থেকে ডেটা asynchronously পড়ে।

Step 5: MongoDB-তে ডাটা ইনসার্ট ও রিড

MongoDB ইনসার্ট (Create Item)

একটি নতুন আইটেম তৈরি করতে:

POST /items/

{
  "name": "Laptop",
  "description": "A powerful gaming laptop",
  "price": 1500.00,
  "tax": 100.00
}

রেসপন্স:

{
  "item_id": "60c72b2f9b1e8d35f8b0f4f9"
}

MongoDB থেকে ডাটা রিড (Get All Items)

GET /items/

রেসপন্স:

[
  {
    "name": "Laptop",
    "description": "A powerful gaming laptop",
    "price": 1500.00,
    "tax": 100.00
  }
]

MongoDB থেকে নির্দিষ্ট আইটেম রিড (Get Item by ID)

GET /items/60c72b2f9b1e8d35f8b0f4f9

রেসপন্স:

{
  "name": "Laptop",
  "description": "A powerful gaming laptop",
  "price": 1500.00,
  "tax": 100.00
}

Step 6: FastAPI অ্যাপ চালানো

FastAPI অ্যাপ্লিকেশন চালানোর জন্য uvicorn ব্যবহার করুন:

uvicorn app:app --reload

এখন আপনি ব্রাউজারে গিয়ে http://127.0.0.1:8000 এ আপনার FastAPI অ্যাপ্লিকেশন দেখতে পারবেন এবং MongoDB-এর সাথে ইন্টিগ্রেটেড CRUD অপারেশন পরীক্ষা করতে পারবেন।

FastAPI এবং MongoDB-এর ইন্টিগ্রেশন খুবই শক্তিশালী এবং সহজ। Motor ড্রাইভার ব্যবহারে MongoDB-তে অ্যাসিঙ্ক্রোনাস কোড ব্যবহার করা সম্ভব, যা স্কেলেবিলিটি এবং পারফরম্যান্স উন্নত করে। MongoDB একটি নমনীয় এবং স্কেলেবল ডাটাবেস হওয়ায় FastAPI-র সাথে এটি একটি চমৎকার সংমিশ্রণ তৈরি করে।

FastAPI এবং MongoDB-এর ইন্টিগ্রেশন খুবই সহজ এবং কার্যকরী। MongoDB একটি NoSQL ডাটাবেস যা JSON-স্টাইল ডকুমেন্ট ব্যবহার করে ডাটা সংরক্ষণ করে। FastAPI-র সাথে MongoDB ব্যবহার করতে Motor লাইব্রেরি ব্যবহার করা হয়, যা MongoDB এর জন্য একটি অ্যাসিঙ্ক্রোনাস ড্রাইভার। এটি FastAPI তে MongoDB এর সাথে ইন্টিগ্রেশন করতে সহায়তা করে এবং অ্যাসিঙ্ক্রোনাস কোড পরিচালনা করতে পারে।

Step 1: MongoDB ইনস্টলেশন

MongoDB ব্যবহার করার জন্য প্রথমে motor লাইব্রেরি ইনস্টল করতে হবে। Motor হলো MongoDB এর অ্যাসিঙ্ক্রোনাস ড্রাইভার, যা asyncio এর উপর ভিত্তি করে কাজ করে।

pip install motor

এছাড়া, MongoDB সার্ভার আপনার সিস্টেমে ইনস্টল থাকতে হবে অথবা আপনি MongoDB এর একটি ক্লাউড সার্ভিস (যেমন MongoDB Atlas) ব্যবহার করতে পারেন।

Step 2: FastAPI এবং MongoDB কনফিগারেশন

FastAPI এবং MongoDB এর ইন্টিগ্রেশন করতে হলে, প্রথমে MongoDB এর সাথে সংযোগ স্থাপন করতে হবে। এখানে আমরা MongoDB এর Motor ড্রাইভার ব্যবহার করব।

উদাহরণ: MongoDB কনফিগারেশন FastAPI তে

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel
from typing import List

app = FastAPI()

# MongoDB Client কনফিগারেশন
@app.on_event("startup")
async def startup_db():
    app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
    app.database = app.mongodb_client.test_db  # MongoDB Database Name

@app.on_event("shutdown")
async def shutdown_db():
    app.mongodb_client.close()

# Pydantic Model
class Item(BaseModel):
    name: str
    description: str

# MongoDB Collection থেকে ডাটা রিড করার ফাংশন
@app.get("/items/", response_model=List[Item])
async def get_items():
    items_cursor = app.database.items.find()  # MongoDB Collection
    items = await items_cursor.to_list(length=100)
    return items

# MongoDB Collection-এ ডাটা ইনসার্ট করার ফাংশন
@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    result = await app.database.items.insert_one(item_dict)  # MongoDB Collection Insert
    return {"id": str(result.inserted_id), **item_dict}

ব্যাখ্যা:

• MongoDB সংযোগ (Connection): AsyncIOMotorClient ব্যবহার করে MongoDB সার্ভারের সাথে অ্যাসিঙ্ক্রোনাস সংযোগ তৈরি করা হয়েছে।
• startup_db এবং shutdown_db: অ্যাপ্লিকেশন শুরু এবং শেষ হওয়ার সময় MongoDB সংযোগ তৈরি ও বন্ধ করতে ব্যবহৃত হচ্ছে।
• Item মডেল: Pydantic মডেল যা MongoDB থেকে ডাটা ইনপুট এবং আউটপুটের জন্য ব্যবহৃত হয়।
• get_items: MongoDB থেকে ডাটা রিড করার ফাংশন।
• create_item: MongoDB-তে ডাটা ইনসার্ট করার ফাংশন।

Step 3: MongoDB-তে ডাটা ইনসার্ট এবং রিড করা

FastAPI অ্যাপ্লিকেশন তৈরি করার পর আপনি MongoDB-তে ডাটা ইনসার্ট এবং রিড করতে পারবেন।

ডাটা ইনসার্ট করার জন্য:

POST /items/

Body:

{
  "name": "Laptop",
  "description": "A high-performance laptop"
}

রেসপন্স:

{
  "id": "605c72ef153207232b0d1985",
  "name": "Laptop",
  "description": "A high-performance laptop"
}

MongoDB থেকে ডাটা রিড করার জন্য:

GET /items/

রেসপন্স:

[
  {
    "id": "605c72ef153207232b0d1985",
    "name": "Laptop",
    "description": "A high-performance laptop"
  },
  {
    "id": "605c72ef153207232b0d1986",
    "name": "Smartphone",
    "description": "A powerful smartphone"
  }
]

এখানে, আমরা MongoDB থেকে সমস্ত items সংগ্রহ করে JSON আকারে রিটার্ন করেছি।

Step 4: MongoDB-এর সাথে অ্যাসিঙ্ক্রোনাস কনফিগারেশন

FastAPI এবং Motor ব্যবহারের ফলে, MongoDB এর সাথে অ্যাসিঙ্ক্রোনাস কোড পরিচালনা করা সম্ভব। MongoDB-এর সাথে ডাটাবেস অপারেশনগুলি অ্যাসিঙ্ক্রোনাসভাবে করা হয়, যাতে অ্যাপ্লিকেশনটি দ্রুত এবং স্কেলেবল হয়।

উদাহরণ: অ্যাসিঙ্ক্রোনাস ডাটা রিড এবং আপডেট

@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
    result = await app.database.items.update_one(
        {"_id": item_id},
        {"$set": item.dict()}
    )
    if result.modified_count == 1:
        return {"message": "Item updated successfully"}
    return {"message": "Item not found"}

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    result = await app.database.items.delete_one({"_id": item_id})
    if result.deleted_count == 1:
        return {"message": "Item deleted successfully"}
    return {"message": "Item not found"}

এখানে:

• update_item: MongoDB ডকুমেন্ট আপডেট করা।
• delete_item: MongoDB ডকুমেন্ট মুছে ফেলা।

Step 5: MongoDB Atlas এর সাথে সংযোগ (Cloud Database)

MongoDB Atlas একটি ক্লাউড-ভিত্তিক MongoDB ডাটাবেস পরিষেবা। আপনি MongoDB Atlas ব্যবহার করে MongoDB-এর ক্লাউড ইনস্ট্যান্সের সাথে সংযোগ করতে পারেন।

MongoDB Atlas এ একটি ডাটাবেস তৈরি করে তার Connection String ব্যবহার করে FastAPI-র সাথে MongoDB সংযোগ করতে পারবেন।

উদাহরণ: MongoDB Atlas কনফিগারেশন

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()

@app.on_event("startup")
async def startup_db():
    app.mongodb_client = AsyncIOMotorClient("your_mongodb_atlas_connection_string")
    app.database = app.mongodb_client.test_db

@app.on_event("shutdown")
async def shutdown_db():
    app.mongodb_client.close()

এখানে your_mongodb_atlas_connection_string MongoDB Atlas থেকে প্রাপ্ত সংযোগ স্ট্রিং।

FastAPI এবং MongoDB (Motor ড্রাইভার ব্যবহার করে) এর ইন্টিগ্রেশন খুবই সহজ এবং কার্যকরী। আপনি অ্যাসিঙ্ক্রোনাস কোড ব্যবহার করে MongoDB ডাটাবেসের সাথে দ্রুত এবং স্কেলেবল অ্যাপ্লিকেশন তৈরি করতে পারেন। MongoDB এবং FastAPI এর এই ইন্টিগ্রেশন ব্যবহারের মাধ্যমে আপনি সহজে ডাটা সংগ্রহ, ইনসার্ট, আপডেট, এবং ডিলিট অপারেশনগুলো পরিচালনা করতে পারবেন।

FastAPI তে Motor লাইব্রেরি ব্যবহার করে MongoDB-এর সাথে asynchronous ইন্টারঅ্যাকশন খুবই সহজ। Motor হলো MongoDB এর জন্য একটি asynchronous ড্রাইভার যা Tornado বা asyncio এ কাজ করতে পারে এবং FastAPI-তে MongoDB অ্যাপ্লিকেশন পরিচালনার জন্য আদর্শ।

এখানে দেখানো হবে কিভাবে Motor লাইব্রেরি ব্যবহার করে MongoDB এর সাথে asynchronous ইন্টারঅ্যাকশন করা যায় এবং FastAPI অ্যাপ্লিকেশনের মধ্যে MongoDB এর সাথে ডেটা ইন্টিগ্রেশন করা যায়।

Step 1: Motor এবং MongoDB সেটআপ

প্রথমে, আপনার প্রকল্পে Motor লাইব্রেরি ইনস্টল করতে হবে। এটি MongoDB এর asynchronous ক্লায়েন্ট ড্রাইভার।

Motor ইনস্টলেশন

pip install motor

এছাড়া, MongoDB সার্ভার চালু থাকতে হবে। আপনি MongoDB ডাউনলোড করে লোকালহোস্টে MongoDB চালু করতে পারেন অথবা MongoDB Atlas ব্যবহার করতে পারেন।

Step 2: FastAPI অ্যাপ্লিকেশন সেটআপ

Motor ব্যবহার করে MongoDB এর সাথে সংযোগ স্থাপন করতে, প্রথমে FastAPI অ্যাপ্লিকেশন তৈরি করতে হবে।

উদাহরণ: MongoDB সেটআপ এবং Motor ব্যবহার

from fastapi import FastAPI, HTTPException
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel
from typing import List

app = FastAPI()

# MongoDB client setup
@app.on_event("startup")
async def startup_db():
    app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
    app.mongodb_db = app.mongodb_client.mydatabase

@app.on_event("shutdown")
async def shutdown_db():
    app.mongodb_client.close()

# Pydantic model for request body
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    collection = app.mongodb_db.items
    result = await collection.insert_one(item.dict())
    item_in_db = await collection.find_one({"_id": result.inserted_id})
    return item_in_db

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: str):
    collection = app.mongodb_db.items
    item = await collection.find_one({"_id": item_id})
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

এখানে:

• AsyncIOMotorClient: এটি MongoDB-এর asynchronous ড্রাইভার।
• on_event("startup"): অ্যাপ্লিকেশন চালু হওয়ার সময় MongoDB ক্লায়েন্ট এবং ডাটাবেস সংযোগ স্থাপন করা হয়।
• on_event("shutdown"): অ্যাপ্লিকেশন বন্ধ হওয়ার সময় MongoDB ক্লায়েন্ট বন্ধ করা হয়।

এছাড়া, Item নামের একটি Pydantic মডেল তৈরি করা হয়েছে যা MongoDB ডাটাবেসে item ডাটা সঞ্চয় করবে।

Step 3: MongoDB Interaction (Insert, Retrieve)

এখন আমরা দেখব কিভাবে MongoDB তে ডাটা ইনসার্ট করা এবং রিট্রিভ করা যায়।

১. Item ইনসার্ট করা (POST রিকোয়েস্ট)

MongoDB তে নতুন item ইনসার্ট করার জন্য, POST রিকোয়েস্ট পাঠাতে হবে।

রিকোয়েস্ট:

POST /items/

{
  "name": "Laptop",
  "description": "A high-end gaming laptop",
  "price": 1500.00,
  "tax": 100.00
}

এটি MongoDB তে একটি নতুন item সংরক্ষণ করবে।

২. Item রিট্রিভ করা (GET রিকোয়েস্ট)

MongoDB তে সংরক্ষিত item গুলি রিট্রিভ করার জন্য GET রিকোয়েস্ট ব্যবহার করা হবে।

রিকোয়েস্ট:

GET /items/{item_id}

এখানে, {item_id} MongoDB ডাটাবেসে সঞ্চিত আইটেমের _id হতে হবে।

Step 4: MongoDB Query Operations

Motor লাইব্রেরি MongoDB তে বিভিন্ন ধরণের query operations সাপোর্ট করে। যেমন:

• find: ডাটাবেস থেকে ডাটা অনুসন্ধান।
• update: ডাটাবেসে ডাটা আপডেট করা।
• delete: ডাটাবেস থেকে ডাটা মুছে ফেলা।

উদাহরণ: MongoDB Query Operations

@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
    collection = app.mongodb_db.items
    update_result = await collection.update_one(
        {"_id": item_id}, {"$set": item.dict()}
    )
    if update_result.modified_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item updated successfully"}

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    collection = app.mongodb_db.items
    delete_result = await collection.delete_one({"_id": item_id})
    if delete_result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

এখানে:

• update_one: MongoDB তে একটি ডকুমেন্ট আপডেট করতে ব্যবহৃত হয়।
• delete_one: MongoDB থেকে একটি ডকুমেন্ট মুছে ফেলার জন্য ব্যবহৃত হয়।

Step 5: Run the FastAPI Application

এখন FastAPI অ্যাপ্লিকেশনটি চালানোর জন্য Uvicorn ব্যবহার করা হবে:

uvicorn main:app --reload

এটি localhost:8000 এ অ্যাপ্লিকেশন চালু করবে।

Step 6: Test the Application

1. POST Request: /items/ এ POST রিকোয়েস্ট পাঠান এবং একটি নতুন item ইনসার্ট করুন।
2. GET Request: /items/{item_id} তে GET রিকোয়েস্ট পাঠান এবং ইনসার্ট করা item টি রিট্রিভ করুন।
3. PUT Request: /items/{item_id} এ PUT রিকোয়েস্ট পাঠান এবং item টি আপডেট করুন।
4. DELETE Request: /items/{item_id} এ DELETE রিকোয়েস্ট পাঠান এবং item টি মুছে ফেলুন।

FastAPI এবং Motor লাইব্রেরি ব্যবহার করে আপনি MongoDB-এর সাথে asynchronous ইন্টারঅ্যাকশন সহজেই করতে পারেন। MongoDB তে ডাটা ইনসার্ট, আপডেট, ডিলিট এবং ফেচ করার কাজগুলি asynchronous ভাবে সম্পন্ন হয়, যা আপনার অ্যাপ্লিকেশনের পারফরম্যান্স বৃদ্ধি করে। Motor MongoDB-এর জন্য একটি শক্তিশালী ড্রাইভার যা asyncio এবং Tornado ভিত্তিক অ্যাপ্লিকেশনগুলির সাথে সম্পূর্ণরূপে কাজ করতে সক্ষম।

MongoDB হলো একটি NoSQL ডেটাবেস যা ডেটা সঞ্চয় এবং পরিচালনার জন্য ব্যবহার করা হয়। FastAPI এবং MongoDB একসাথে ব্যবহার করে আপনি ডেটাবেসে CRUD (Create, Read, Update, Delete) অপারেশন সম্পাদন করতে পারেন। FastAPI-র সাথে MongoDB ব্যবহারের জন্য আমরা motor লাইব্রেরি ব্যবহার করব, যা MongoDB-এর জন্য একটি অ্যাসিঙ্ক্রোনাস ড্রাইভার।

এই টিউটোরিয়ালে আমরা দেখব কীভাবে FastAPI এবং MongoDB এর মাধ্যমে CRUD অপারেশন সম্পাদন করা যায়।

Step 1: প্রয়োজনীয় লাইব্রেরি ইনস্টলেশন

প্রথমে, MongoDB ব্যবহার করতে motor লাইব্রেরি এবং FastAPI ইনস্টল করতে হবে:

pip install fastapi
pip install uvicorn
pip install motor

Step 2: MongoDB এর সাথে সংযোগ স্থাপন

MongoDB-তে সংযোগ স্থাপন করতে motor ব্যবহার করা হবে, যা একটি অ্যাসিঙ্ক্রোনাস ড্রাইভার।

MongoDB সংযোগ কনফিগারেশন:

from motor.motor_asyncio import AsyncIOMotorClient
from fastapi import FastAPI

app = FastAPI()

# MongoDB ডাটাবেসের সংযোগ
@app.on_event("startup")
async def startup_db():
    app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")  # MongoDB URL
    app.db = app.mongodb_client["mydatabase"]  # ডাটাবেসের নাম
    print("Connected to MongoDB")

@app.on_event("shutdown")
async def shutdown_db():
    app.mongodb_client.close()
    print("MongoDB connection closed")

এখানে:

• AsyncIOMotorClient: MongoDB এর জন্য অ্যাসিঙ্ক্রোনাস ক্লায়েন্ট তৈরি করা হয়েছে।
• app.db: MongoDB ডাটাবেসের সংযোগ।
• on_event("startup"): অ্যাপের শুরুতে MongoDB সংযোগ করা হবে।
• on_event("shutdown"): অ্যাপ বন্ধ করার সময় MongoDB সংযোগ বন্ধ হবে।

Step 3: MongoDB Collection এর সাথে CRUD অপারেশন

1. Create (ডেটা তৈরি করা)

MongoDB তে নতুন ডকুমেন্ট তৈরি করতে insert_one() ব্যবহার করা হয়।

from pydantic import BaseModel
from fastapi import HTTPException, status

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    item_dict = item.dict()
    result = await app.db["items"].insert_one(item_dict)  # "items" collection এ ডেটা সংরক্ষণ
    item_dict["_id"] = str(result.inserted_id)  # inserted_id MongoDB থেকে ফেরত আসে
    return item_dict

এখানে:

• insert_one(item_dict): নতুন ডকুমেন্ট items collection এ ইনসার্ট করা হয়।
• inserted_id: MongoDB থেকে ফেরত আসা ইনসার্টেড ডকুমেন্টের আইডি।

2. Read (ডেটা পড়া)

MongoDB তে ডেটা পড়তে find_one() অথবা find() ব্যবহার করা হয়।

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    item = await app.db["items"].find_one({"_id": item_id})  # আইডি অনুসারে ডেটা খোঁজা
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return {"name": item["name"], "description": item.get("description", None), "price": item["price"]}

এখানে:

• find_one(): item_id অনুসারে MongoDB থেকে একটি ডকুমেন্ট খুঁজে পাওয়া।
• item.get("description", None): description ফিল্ডটি যদি না থাকে তবে None ফেরত দেয়।

3. Update (ডেটা আপডেট করা)

MongoDB তে ডেটা আপডেট করতে update_one() ব্যবহার করা হয়।

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    updated_item = await app.db["items"].find_one_and_update(
        {"_id": item_id},
        {"$set": item.dict()},
        return_document=True  # MongoDB-তে আপডেট হওয়া ডকুমেন্ট ফেরত আনবে
    )
    if updated_item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return updated_item

এখানে:

• find_one_and_update(): এটি MongoDB ডকুমেন্টটি খুঁজে পায় এবং আপডেট করে।
• $set: MongoDB আপডেট অপারেশনে নতুন মান সেট করতে ব্যবহৃত হয়।

4. Delete (ডেটা মুছে ফেলা)

MongoDB তে ডেটা মুছতে delete_one() ব্যবহার করা হয়।

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    result = await app.db["items"].delete_one({"_id": item_id})
    if result.deleted_count == 0:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return {"message": "Item deleted successfully"}

এখানে:

• delete_one(): MongoDB থেকে একটি ডকুমেন্ট মুছে ফেলা হয়।
• deleted_count: এটি জানায় যে ডকুমেন্ট কতটি মুছে ফেলা হয়েছে। যদি 0 হয়, তাহলে ডকুমেন্টটি পাওয়া যায়নি।

Step 4: FastAPI অ্যাপ চালানো

uvicorn main:app --reload

এটি আপনার FastAPI অ্যাপকে চালু করবে এবং MongoDB-এর সাথে সংযোগ স্থাপন করবে।

MongoDB এর সাথে CRUD অপারেশন এর সম্পূর্ণ উদাহরণ:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from motor.motor_asyncio import AsyncIOMotorClient
from bson import ObjectId

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

@app.on_event("startup")
async def startup_db():
    app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
    app.db = app.mongodb_client["mydatabase"]

@app.on_event("shutdown")
async def shutdown_db():
    app.mongodb_client.close()

# Create Item
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    item_dict = item.dict()
    result = await app.db["items"].insert_one(item_dict)
    item_dict["_id"] = str(result.inserted_id)
    return item_dict

# Read Item
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    item = await app.db["items"].find_one({"_id": ObjectId(item_id)})
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    item["_id"] = str(item["_id"])
    return item

# Update Item
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    updated_item = await app.db["items"].find_one_and_update(
        {"_id": ObjectId(item_id)},
        {"$set": item.dict()},
        return_document=True
    )
    if updated_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    updated_item["_id"] = str(updated_item["_id"])
    return updated_item

# Delete Item
@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    result = await app.db["items"].delete_one({"_id": ObjectId(item_id)})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

FastAPI এবং MongoDB একসাথে ব্যবহার করে আপনি সহজেই CRUD অপারেশন সম্পাদন করতে পারবেন। MongoDB এর অ্যাসিঙ্ক্রোনাস ড্রাইভার motor ব্যবহার করে FastAPI-র মাধ্যমে ডেটাবেসের সাথে ইন্টারঅ্যাক্ট করা সম্ভব। এটি ডেটা ইনসার্ট, আপডেট, ডিলিট, এবং রিড করার প্রক্রিয়াকে দ্রুত এবং স্কেলেবল করে তোলে।

FastAPI তে MongoDB ব্যবহার করার জন্য আমরা Pydantic Models এবং Motor (MongoDB-এর জন্য অ্যাসিঙ্ক্রোনাস ড্রাইভার) ব্যবহার করে সহজেই ডাটাবেসের সাথে ইন্টিগ্রেশন করতে পারি। এখানে আমরা দেখব কিভাবে FastAPI তে MongoDB সংযোগ স্থাপন করতে হয় এবং MongoDB ডকুমেন্টগুলি Pydantic Model এর মাধ্যমে কিভাবে পরিচালনা করা যায়।

Step 1: প্রয়োজনীয় লাইব্রেরি ইনস্টল করা

MongoDB এর সাথে কাজ করার জন্য আপনাকে Motor এবং Pydantic লাইব্রেরি ইনস্টল করতে হবে।

pip install motor pydantic fastapi

Motor MongoDB এর জন্য একটি অ্যাসিঙ্ক্রোনাস ড্রাইভার যা FastAPI-এর সাথে খুব ভালোভাবে কাজ করে।

Step 2: MongoDB সংযোগ স্থাপন করা

MongoDB সংযোগ স্থাপন করতে Motor ব্যবহার করা হয়। প্রথমে MongoDB এর URI (যেমন mongodb://localhost:27017) ব্যবহার করে ডাটাবেসের সাথে সংযোগ স্থাপন করতে হবে।

উদাহরণ: MongoDB সংযোগ কনফিগারেশন

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel
from typing import List

app = FastAPI()

# MongoDB URI এবং ডাটাবেস কনফিগারেশন
MONGO_DETAILS = "mongodb://localhost:27017"  # MongoDB URI
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client.test_db  # MongoDB ডাটাবেস
collection = database.items  # MongoDB কালেকশন

# Pydantic Model
class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float | None = None

এখানে:

• Motor Client ব্যবহার করে MongoDB এর সাথে সংযোগ স্থাপন করা হয়েছে।
• test_db MongoDB ডাটাবেস এবং items MongoDB কালেকশন তৈরি করা হয়েছে।

Step 3: Pydantic Model এবং MongoDB Document Integration

FastAPI তে MongoDB ডেটা ইন্টিগ্রেট করার জন্য Pydantic Model ব্যবহার করা হয়, যার মাধ্যমে ডাটা যাচাই এবং ডাটা ট্রান্সফার করা হয়। MongoDB ডকুমেন্ট গুলি Pydantic Model এর মাধ্যমে ম্যানিপুলেট করা হয়।

উদাহরণ: Pydantic Model এবং MongoDB ডকুমেন্ট হ্যান্ডলিং

from motor.motor_asyncio import AsyncIOMotorClient
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

MONGO_DETAILS = "mongodb://localhost:27017"
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client.test_db
collection = database.items

# Pydantic Model
class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float | None = None

# MongoDB Document হিসেবে Mongodb এর ডাটা পাওয়া
def item_helper(item) -> dict:
    return {
        "name": item["name"],
        "description": item["description"],
        "price": item["price"],
        "tax": item["tax"],
    }

# একটি নতুন item ডকুমেন্ট MongoDB তে ইনসার্ট করা
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    item_data = item.dict()  # Pydantic Model থেকে dict কনভার্ট
    result = await collection.insert_one(item_data)  # MongoDB তে ডাটা ইনসার্ট
    created_item = await collection.find_one({"_id": result.inserted_id})
    return item_helper(created_item)  # MongoDB ডকুমেন্টকে Pydantic Model হিসেবে ফেরত দেয়া

এখানে:

• item_helper() ফাংশনটি MongoDB ডকুমেন্টকে Pydantic Model ডেটাতে কনভার্ট করে।
• create_item() রাউটটি একটি নতুন item MongoDB ডাটাবেসে ইনসার্ট করে এবং সেই ডকুমেন্টটি Pydantic Model হিসেবে ফেরত দেয়।

Step 4: MongoDB থেকে ডাটা রিট্রিভ করা

MongoDB থেকে ডাটা রিট্রিভ করতে এবং সেই ডাটাকে Pydantic Model এর সাথে কনভার্ট করার জন্য find() মেথড ব্যবহার করা হয়।

উদাহরণ: MongoDB থেকে ডাটা রিট্রিভ করা

@app.get("/items/", response_model=List[Item])
async def get_items():
    items = []
    async for item in collection.find():  # MongoDB থেকে ডাটা রিট্রিভ
        items.append(item_helper(item))  # MongoDB ডকুমেন্টকে Pydantic Model হিসেবে কনভার্ট
    return items

এখানে:

• find() ব্যবহার করে MongoDB থেকে সব ডকুমেন্ট রিট্রিভ করা হয়েছে।
• item_helper() ফাংশনটি MongoDB ডকুমেন্টকে Pydantic Model এর সাথে কনভার্ট করে।

রিকোয়েস্ট উদাহরণ:

GET /items/

রেসপন্স:

[
  {
    "name": "Laptop",
    "description": "A high-performance laptop",
    "price": 1500.00,
    "tax": 120.00
  },
  {
    "name": "Phone",
    "description": "A modern smartphone",
    "price": 799.99,
    "tax": 60.00
  }
]

Step 5: MongoDB ডকুমেন্ট আপডেট করা

MongoDB ডকুমেন্ট আপডেট করার জন্য update_one() বা update_many() মেথড ব্যবহার করা হয়।

উদাহরণ: MongoDB ডকুমেন্ট আপডেট করা

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    updated_item = await collection.update_one(
        {"_id": item_id}, {"$set": item.dict()}  # MongoDB ডকুমেন্ট আপডেট
    )
    if updated_item.matched_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

এখানে:

• update_one() ব্যবহার করে MongoDB ডকুমেন্ট আপডেট করা হয়েছে।

Step 6: MongoDB ডকুমেন্ট মুছে ফেলা

MongoDB ডকুমেন্ট মুছে ফেলার জন্য delete_one() মেথড ব্যবহার করা হয়।

উদাহরণ: MongoDB ডকুমেন্ট মুছে ফেলা

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    deleted_item = await collection.delete_one({"_id": item_id})  # MongoDB ডকুমেন্ট মুছে ফেলা
    if deleted_item.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

এখানে:

• delete_one() ব্যবহার করে MongoDB ডকুমেন্ট মুছে ফেলা হয়েছে।

FastAPI তে Pydantic Models ব্যবহার করে MongoDB এর ডকুমেন্ট হ্যান্ডল করা খুবই সহজ এবং কার্যকর। MongoDB-এ ডাটা সংরক্ষণ, রিট্রিভ, আপডেট এবং মুছে ফেলার জন্য FastAPI ও Pydantic মডেল ব্যবহার করা যেতে পারে। Motor লাইব্রেরি MongoDB-এর অ্যাসিঙ্ক্রোনাস ড্রাইভার হিসেবে ব্যবহৃত হয়, যা FastAPI এর সাথে খুব ভালোভাবে কাজ করে।